package com.qidian.buy.search.receives;

import com.qidian.buy.dto.ItemMessageDTO;
import com.qidian.buy.search.service.SearchItemService;
import com.qidian.buy.util.ThreadPoolTools;
import com.qidian.buy.util.Tools;
import com.qidian.buy.util.pojo.QdResult;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Author: 戴着假发的程序员
 * Date: 2024-05-15 14:25
 * company: 起点编程
 **/
@Component
public class ItemIdReceive implements ChannelAwareMessageListener {
    @Autowired
    private SearchItemService searchItemService;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        // 获取消息的唯一性ID
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        ItemMessageDTO dto = null;
        try {
            String msg = new String(message.getBody(),"UTF-8");
            // 取出消息
            dto = Tools.json2object(msg, ItemMessageDTO.class);
            // 只要拿到消息就立刻确认, 防止消息被其他的节点重复消息
            // 手动确认
            channel.basicAck(deliveryTag,true);
            QdResult result = QdResult.error();
            if (dto != null){
                Long itemId = dto.getItemId();
                if (dto.getOption().equals(ItemMessageDTO.OP_INSERT)){
                    // 插入
                    result = searchItemService.importIndex(itemId);
                }else if (dto.getOption().equals(ItemMessageDTO.OP_UPDATE)){
                    // 更新
                }else if (dto.getOption().equals(ItemMessageDTO.OP_DELETE)){
                    // 删除
                }
            }
            if(result.getCode() != 200){
                throw new RuntimeException("同步索引失败");
            }
        }catch (Exception e){
            if(dto == null){
                channel.basicReject(deliveryTag, false);
            }else{
                final ItemMessageDTO tempDto = dto;
                // 开启一个子线程进行重试
                ThreadPoolExecutor pool = ThreadPoolTools.getPool("SEARCH_SYNC_THREAD");
                pool.execute(()->{
                    QdResult result = QdResult.error();
                    // 重试5次
                    for (int i = 0;i < 5;i ++){
                        try {
                            if(tempDto.getOption().equals(ItemMessageDTO.OP_INSERT)){
                                // 插入
                                result = searchItemService.importIndex(tempDto.getItemId());
                            }else if (tempDto.getOption().equals(ItemMessageDTO.OP_UPDATE)){
                                // 更新
                            }else if (tempDto.getOption().equals(ItemMessageDTO.OP_DELETE)){
                                // 删除
                            }
                            if(result.getCode() == 200){
                                break;
                            }
                            TimeUnit.SECONDS.sleep(1);// 1秒钟重试一次
                        } catch (Exception ex) {
                            //throw new RuntimeException(ex);
                            // 如果出现异常继续重试
                        }
                    }
                    // 循环结束之后，判断是否已经成功了
                    if(result.getCode() != 200){
                        // 记录失败的信息
                        stringRedisTemplate.opsForList().leftPush("SEARCH_SYNC_FAL",Tools.object2json(tempDto));
                    }
                });
            }
        }
    }
}
